package my.thesis.bolts.apicalls;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.HttpURLConnection;
import java.net.MalformedURLException;
import java.net.URL;
import java.net.URLConnection;
import java.util.Map;

import org.json.JSONArray;
import org.json.JSONException;
import org.json.JSONObject;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class CountryDetectorAsync extends BaseRichBolt
{
	private static final long serialVersionUID = 1L;
	private static final String MAPQUESTKEY = "Fmjtd%7Cluur256tng%2Cas%3Do5-9w72lz";
	private static final boolean callapi=true;
	private OutputCollector collector;
	
	private static String result=null;

	public void prepare(Map stormConf, TopologyContext context,OutputCollector collector) 
	{
		this.collector=collector;
	}
	
	/*
	 A BOLT CAN EMIT A TUPLE AT ALMOST ANY TIME GRANTED THAT YOU HAVE THE OUTPUT COLLECTOR
	 A GOOD APPROACH IS TO MAKE THE REST CALL ASYNCHRONOUSLY IN A SEPERATE THREAD THEN HAVE THAT THREAD EMIT THE RESULTS AND CLOSE
	 THAT WAY THE TOPOLOGY CAN KEEP RUNNING
	 LEEI O ENAS
	 Storm requires that the emit method on the OutputCollector (used for emitting tuples from one task to another) must always be called from the same thread, 
	 otherwise NullPointerExceptions will occur from within Storm itself.
	 ALLA KAI ALLOU
	 "Tuples can be emitted at anytime from the bolt -- in the prepare, execute, or cleanup methods, or even asynchronously in another thread"
	 */
	public void execute(Tuple tuple) 
	{
		if(callapi)
		{

			final double lat = tuple.getDoubleByField("lat");
			final double lon = tuple.getDoubleByField("lon");
			final Long id = tuple.getLongByField("id");
			final String text=tuple.getStringByField("text");
						
		    if(lat != 0.0 && lon != 0.0)
			{
		    	new Thread(new Runnable() 
			    {
			    	public void run() 
			    	{
					    	result= GetCountryByLatLon(lat,lon);
					    		
/*							if(res!=null)
							{
								result=res;
								
								//EMIT FROM OTHER THREAD
								collector.emit(new Values(id,result,text));
							}
							else
							{
								 System.out.println("Country not found...");
							}*/
			    	}
			    }).start();
			    
		    	//EMIT FROM MAIN THREAD
				if(result!=null)
				{
					collector.emit(new Values(id,result,text));
				}
				else
				{
					 System.out.println("Country not found...");
				}
			}
		}
		else
		{
			double id=tuple.getDoubleByField("id");
			String text=tuple.getStringByField("text");
			String country=tuple.getStringByField("country");
			collector.emit(new Values(id,country,text));
			
			//RELIABLE
			//collector.emit(tuple,new Values(id,country,text));
			
			//RELIABLE
			//collector.ack(tuple);
		}
	}	
	
	public void declareOutputFields(OutputFieldsDeclarer declarer) 
	{
		declarer.declare(new Fields("id","country","text"));		
	}
		
	//BOOK TWITTER DATA ANALYTICS - ΑΥΤΟ ΤΟ ΕΧΩ ΠΡΟΣΑΡΜΟΣΕΙ ΑΡΚΕΤΑ
    /**
     * Translates a location string to coordinates using the database or Nominatim Service
     * @param loc
     * @return
     */
	public String GetCountryByLatLon(Double lat,Double lon)
	{
		 //test browser -- IT WORKS!
		 //http://open.mapquestapi.com/geocoding/v1/reverse?key=Fmjtd%7Cluur256tng%2Cas%3Do5-9w72lz&callback=renderReverse&location=29.897126,-95.694483
		
		 String url="http://open.mapquestapi.com/geocoding/v1/reverse?key="+MAPQUESTKEY+"&callback=renderReverse&location="+lat+","+lon+"&outFormat=json";
		 String page = ReadHTML(url);
		 
         if(page!=null)
         {
             try
             {
            	page = page.substring(page.lastIndexOf("(") + 1);
            	page = page.substring(0, page.indexOf(')'));

            	JSONObject pageJson = new JSONObject(page);
            	JSONArray results = pageJson.getJSONArray("results");
            	
            	JSONObject locationsObj = results.getJSONObject(0);
            	JSONArray locationsArr = locationsObj.getJSONArray("locations");
            	
            	if(locationsArr.length()>0)
                {             	   
            	   String country = new String(locationsArr.getJSONObject(0).getString("adminArea1"));
            	   
            	   System.out.println("Found country: " + country);
            	   
            	   return country;
                }
            	
            	
             }
             catch(JSONException ex)
             {
                System.out.println("Something went terribly wrong...");
             }
         }
         
         return null;
	}
		
	//BOOK TWITTER DATA ANALYTICS
    /**
     * Extracts the html content of a URL
     * @param url
     * @return html page
     */
	public String ReadHTML(String url)
    {
        URLConnection conn = null;
        URL theURL = null;
        try
        {
                theURL = new URL(url);
        }
        catch ( MalformedURLException e)
        {
                System.out.println("Bad URL: " + theURL);
                return null;
        }
        String page = "";
        try
        {
        	//Returns a URLConnection instance that represents a connection to the remote object referred to by the URL.
        	//It should be noted that a URLConnection instance does not establish the actual network connection on creation. 
        	//This will happen only when calling URLConnection.connect().
        	//If for the URL's protocol (such as HTTP or JAR), there exists a public, specialized URLConnection subclass belonging to one of the following packages or one 
        	//of their subpackages: java.lang, java.io, java.util, java.net, the connection returned will be of that subclass. For example, for HTTP an HttpURLConnection will 
        	//be returned, and for JAR a JarURLConnection will be returned.
        	
            conn = theURL.openConnection();
            HttpURLConnection huc = (HttpURLConnection) conn;
            conn.setConnectTimeout(2000);
          
            conn.connect();
            
            if(huc.getResponseCode()>=400 && huc.getResponseCode()<=404)
            {
                 return null;
            }
            
            BufferedReader bRead = new BufferedReader(new InputStreamReader((InputStream) conn.getContent()));
            String temp=null;
            while( (temp= bRead.readLine())!=null)
              {
                  page = page+"\n"+temp;
              }
              bRead.close();
        }
        catch (IOException e) 
        {
                return null;
        }
        
        return page;
    }
	
}
